//服务器业务类cpp

#include "chatservice.hpp"
#include <muduo/base/Logging.h> //muduo日志头文件
#include "hash_salt.hpp"

using namespace muduo;

ChatService* ChatService::getInstance(){
    //C++11线程安全

    //单例对象
    static ChatService _service;
    return &_service;
}

//构造函数注册消息以及对应的Handler回调操作
ChatService::ChatService(){
    //哈希表插入某个消息类型对应的操作回调函数
    //登录消息类型--->login()函数
    _msgHandlerMap.insert(
        {EnMsgType::LOG_MSG,std::bind(&ChatService::login,this,_1,_2,_3)});
    
    //注册消息类型--->reg()
    _msgHandlerMap.insert(
        {EnMsgType::REG_MSG,std::bind(&ChatService::reg,this,_1,_2,_3)});

    _msgHandlerMap.insert(
        {EnMsgType::ONE_CHAT_MSG,std::bind(&ChatService::oneChat,this,_1,_2,_3)});

    _msgHandlerMap.insert(
        {EnMsgType::ADD_FRIEND_MSG,std::bind(&ChatService::addFriend,this,_1,_2,_3)});


    //群组相关业务绑定

    _msgHandlerMap.insert(
        {EnMsgType::CREATE_GROUP_MSG,std::bind(&ChatService::createGroup,this,_1,_2,_3)});

    _msgHandlerMap.insert(
        {EnMsgType::ADD_GROUP_MSG,std::bind(&ChatService::addGroup,this,_1,_2,_3)});

    _msgHandlerMap.insert(
        {EnMsgType::GROUP_CHAT_MSG,std::bind(&ChatService::groupChat,this,_1,_2,_3)});

    //客户端注销业务
    _msgHandlerMap.insert(
        {EnMsgType::LOGINOUT_MSG,std::bind(&ChatService::loginOut,this,_1,_2,_3)});

#ifdef USE_REDIS

    //连接Redis并注册回调函数
    if(_redis.connect()){
        _redis.init_notify_handler(std::bind(&ChatService::handleRedisSubscribeMessage,this,_1,_2));
    }else{
        LOG_ERROR<<"redis conn error!";
    }

#endif

#ifdef USE_KAFKA
    //初始化该kafka消费者生产者，即一台服务器是一个生产者可以转发消息的同时，又是一个消费者，
    //监听登录在当前服务器的用户的消息

    this->_producer=new KafkaProducer("localhost:9092","maintopic",0);
    this->_consumer=new KafkaConsumer("localhost:9092","maintopic","1",0);

    if(!_producer->init()){
        std::cerr<<"producer init failed!\n";
        exit(-1);
    }

    if(!_consumer->init()){
        std::cerr<<"consumer init failed!\n";
        exit(-1);
    }
    _consumer->sub_msg_function(std::bind(&ChatService::handleKafkaSubscribeMessage,this,_1));
    
    //kafka消费者监听线程函数，利用了lambda
    std::thread t_listen([](KafkaConsumer* consum){consum->start(1000);},_consumer);
    t_listen.detach();
    
#endif


}

//获取某个类型的事件的处理器
MsgHandler ChatService::getHandler(EnMsgType msgID){
    auto iter=_msgHandlerMap.find(msgID);
    
    //如果找不到对应的回调函数，则输出日志
    if(iter==_msgHandlerMap.end()){
        //使用lambda表达式，返回空操作的回调函数，并输出错误日志
        return [=](const TcpConnectionPtr& conn,json& js,Timestamp time){
            LOG_ERROR<<"msgID: "<<int(msgID)<<" could not find handler!\n";
        };
            
    }
    return iter->second;
}

ChatService::~ChatService(){

#ifdef USE_KAFKA
    //删除在构建时创造的kafka相关对象指针
    if(_producer){
        delete _producer;
        _producer=nullptr;
    }

    if(_consumer){
        delete _consumer;
        _consumer=nullptr;
    }

#endif
}

//处理登录业务
void ChatService::login(const TcpConnectionPtr& conn,json& js,Timestamp time){
    //LOG_INFO<<"do login() success !\n";
    int id=js["id"].get<int>();
    string pwd=js["password"];

    //通过UserModel类调用数据库代码，返回数据库结果并提取成一个user类对象
    User user=_userModel.query(id);

    //加盐哈希验证
    salt_hash wantpasswd(pwd);//用户输入过来的密码
    wantpasswd.setSalt(user.getSalt());
    
    if(user.getId()!=-1&&user.getPwd()==wantpasswd.getSalt_hash()){
        //账号存在且密码正确
        if(user.getState()=="online"){
            //账号密码正确但是账号已经登录状态
            //不允许重复登录
            json response;
            response["msgid"]=int(EnMsgType::LOG_MSG_ACK);
            response["errno"]=2;    //注意这里的错误码为2，表示想要登录的账号已经登录了
            response["errmsg"]="登录失败：该账号已登录，请重新输入账户！";

            conn->send(response.dump());

        }else{
            //登录成功，账号密码正确，并且账号没有出现给占用的情况
            //将连接成功的连接对象存放
            //注意加锁，这里lock_guard<mutex>相当于锁的智能指针，出了大括号作用域就会析构解锁
            {
                std::lock_guard<std::mutex> lock(_ConnMutx);
                _userConnMap.insert({user.getId(),conn});
            }
            

            //修改登录状态为Online
            user.setState("online");
            bool res = _userModel.updateState(user);    //!有线程安全争议，需要加锁
            if(!res){
                LOG_ERROR<<"userID: "<<user.getId()<<"修改状态出错！";
            }

        #ifdef USE_REDIS
            //登录用户成功，服务器从Redis消息队列订阅该用户的消息
            if(_redis.subscribe(id)){
                LOG_INFO<<"redis订阅成功！";
            }else{
                LOG_ERROR<<"redis订阅失败！";
            }
        #endif //USE_REDIS
            

        #ifdef USE_KAFKA

            //创建用户id的话题，不然的话消费者是订阅不了，导致错误的
            _producer->createTopic("user"+std::to_string(id));
            sleep(1);
            //用户上线，订阅
            _consumer->subscribe("user"+std::to_string(id));
        #endif //USE_KAFKA


            json response;
            //1、添加登录响应消息
            response["msgid"]=int(EnMsgType::LOG_MSG_ACK);
            response["errno"]=0;
            response["id"]=id;
            response["name"]=user.getName();
            LOG_INFO<<"检测到用户登录: "<<user.getName();

            //2、查询该登录用户是否有离线消息，有的话一并发送给刚登录的这个用户
            vector<string> vec=_offlineMessageModel.query(id);
            if(!vec.empty()){
                //有离线消息，则把vector数组放入json中
                response["offlinemsg"]=vec;

                //读取离线消息后，则将该用户所有离线消息从数据库中移除
                _offlineMessageModel.remove(id);
            }
            

            //3、查询好友列表，用一个string数组包装着这些用户的列表
            vector<User> userArr=_friendModel.query(id);
            if(!userArr.empty()){
                vector<string> resArr;
                for(User& usr:userArr){
                    json js;
                    js["id"]=usr.getId();
                    js["name"]=usr.getName();
                    js["state"]=usr.getState();

                    resArr.push_back(js.dump());
                }
                response["friends"]=resArr;
            }

            //4、查询该用户所在的群组信息（!有点难理解的，设计到两个for）
            vector<Group> groupUserArr=_groupmodel.queryGroups(id);
            if(!groupUserArr.empty()){
                //如果有加入的群组则进行转换，也是用string数组将它们包装
                vector<string> groupStrArr;
                for(Group& group:groupUserArr){
                    json grpjson;
                    grpjson["groupid"]=group.getID();
                    grpjson["groupname"]=group.getName();
                    grpjson["groupdesc"]=group.getDesc();

                    //存储当前群组里的成员列表的数组
                    vector<string> usersInEachGroup;
                    for(GroupUser& guser: group.getUsers()){
                        json userJson;
                        userJson["id"]=guser.getId();
                        userJson["name"]=guser.getName();
                        userJson["state"]=guser.getState();
                        userJson["role"]=guser.getRole();
                        
                        usersInEachGroup.push_back(userJson.dump());
                    }
                    //将群里面的成员列表放入json
                    grpjson["users"]=usersInEachGroup;
                    
                    //将grpjson打入到groupStrArr，代表其中一个群的信息遍历完毕
                    groupStrArr.push_back(grpjson.dump());

                }
                //遍历完该用户所在的每一个群组信息之后
                //将封装好的string数组放入响应客户端的Json中
                response["groups"]=groupStrArr;

            }


            conn->send(response.dump());
        }

    }else{
        //登录失败，账户不存在或密码错误
        json response;
        response["msgid"]=int(EnMsgType::LOG_MSG_ACK);
        response["errno"]=1;
        response["errmsg"]="登录失败：用户名不存在或密码错误！";
        
        conn->send(response.dump());
    }



}

//处理注册业务，注册时只用到name和password
void ChatService::reg(const TcpConnectionPtr& conn,json& js,Timestamp time){
    //LOG_INFO<<"do reg() success !\n";
    string name=js["name"];
    string password=js["password"];

    //new: 加盐哈希处理密码
    salt_hash shash(password);
    string salt=shash.randSalt();
    shash.setSalt(salt);
    string passwd_to_hash=shash.getSalt_hash();

    User user;
    user.setName(name);
    user.setPwd(passwd_to_hash);
    user.setSalt(salt);

    //调用user操作类UserModel对象来实现往数据库添加语句，并判断对错
    bool res=_userModel.insert(user);
    if(res){
        //注册成功

        //注册成功响应报文json
        json response;
        response["msgid"]=int(EnMsgType::REG_MSG_ACK);
        response["id"]=user.getId();
        response["errno"]=0;    //注册成功时错误代码为0

        conn->send(response.dump());

    }else{
        //注册失败

        //注册失败响应报文json
        json response;
        response["msgid"]=int(EnMsgType::REG_MSG_ACK);
        response["id"]=user.getId();
        response["errno"]=1;    //注册错误时错误代码为1
        //response["errmsg"]="user registered fail!\n";   //注册错误消息（可选）

        conn->send(response.dump());
    }
    
}

//服务器异常关闭，重置客户业务方法
void ChatService::reset(){
    //将处于online的用户状态改为offline
    _userModel.resetState();
}


//客户端异常关闭处理方案
void ChatService::clientCloseException(const TcpConnectionPtr& conn){
    User user;
    {
        std::lock_guard<std::mutex> lock(_ConnMutx);
        //遍历已登录的连接，查找异常关闭的客户端，并从哈希表中删除它
        for(auto it=_userConnMap.begin();it!=_userConnMap.end();++it){
            if(it->second==conn){
                user.setID(it->first);
                _userConnMap.erase(it);
                break;
            }
        }
    }
    //用户异常关闭，取消订阅
#ifdef USE_REDIS
    //用户异常关闭也等于下线，也要取消Redis中消息队列的订阅
    _redis.unsubscribe(user.getId());
#endif


#ifdef USE_KAFKA
    //kafka取消订阅
    if(!_consumer->unsubscribe("user"+std::to_string(user.getId())))
        cout<<"unsubscribe error! id: "<<user.getId()<<endl;
#endif

    //更新该异常客户端状态为下线状态
    if(user.getId()!=-1){
        user.setState("offline");
        _userModel.updateState(user);
    }

    
    
}

//点对点（用户对用户）聊天业务处理
void ChatService::oneChat(const TcpConnectionPtr& conn,json& js,Timestamp time){

    //printf("one chat call\n");
    //获取信息目标客户端id
    int toid=js["toid"].get<int>();
    

    //查找该id用户是否在线，不在线则转为离线消息
    //这里之所以把发送过程都放在互斥锁作用域内
    //是因为可能在发送时候conn就给删了，所以要保证线程安全
    {
        std::lock_guard<std::mutex> lock(_ConnMutx);
        auto it=_userConnMap.find(toid);
        if(it!=_userConnMap.end()){
            //目标用户在线，直接传输消息
            //这里是调用目标客户端的连接对象，用这个对象向目标客户端发送原封不动的json
            //等于服务器起到的是个转发效果
            it->second->send(js.dump());
            return;
        }

    }

    //在本服务器在线用户列表里找不到该用户，首先查找数据库是否在线
    //数据库显示在线则表明用户在另一个服务器中，这时候就要把消息传到Redis消息队列中
    User user=_userModel.query(toid);
    if(user.getState()=="online"){
#ifdef USE_REDIS
        //在其它服务器在线的话，把消息传到redis消息队列以便达到另一服务器中
        _redis.publish(toid,js.dump());
        return;
#endif//USE_REDIS

#ifdef USE_KAFKA
        //生产者发布消息到kafka消息队列中
        _producer->send("user"+std::to_string(toid),js.dump());
        return;
#endif

    }

    //数据库也不在线则转为离线消息
    _offlineMessageModel.insert(toid,js.dump());

}


//添加好友业务
void ChatService::addFriend(const TcpConnectionPtr& conn,json& js,Timestamp time){
    //获取发送请求的客户id
    int userid=js["id"].get<int>();

    //想添加的好友id
    int friendid=js["friendid"].get<int>();

    if(_friendModel.haveUser(friendid)==false){
        //找不到该好友，返回错误报文给客户端
        json js;
        js["msgid"]=int(EnMsgType::ADD_FRIEND_ACK);
        //规定ADD_FRIEND_ACK中若错误代码为1，则为找不到用户
        js["errno"]=1;  
        js["msg"]="找不到该用户，请重新输入想添加的好友ID！";

        conn->send(js.dump());
    }
    else{//!有缺陷，如果想添加的好友已经存在了，则跳过流程（这里还没写该逻辑代码）
        //找的到用户
        _friendModel.insert(userid,friendid);

        //找到该好友，返回添加成功报文
        json js;
        js["msgid"]=int(EnMsgType::ADD_FRIEND_ACK);
        //规定ADD_FRIEND_ACK中若添加成功，则错误代码为0
        js["errno"]=0;  
        js["msg"]="添加好友成功！";

        conn->send(js.dump());
    }
}



//创建群组业务
void ChatService::createGroup(const TcpConnectionPtr& conn,json& js,Timestamp time){
    //创建者id
    int userid=js["id"].get<int>();
    //群名称
    string groupName=js["groupname"];
    //群备注
    string groupDesc=js["groupdesc"];

    Group group(-1,groupName,groupDesc);

    //创建群组响应报文Json
    json createGroupACK;
    createGroupACK["msgid"]=int(EnMsgType::CREATE_GROUP_ACK);

    //若创建群成功
    if(_groupmodel.createGroup(group)==true){

        //添加userid进群组并设置它为创建者
        _groupmodel.addGroup(userid,group.getID(),"creator");

        //响应json错误码为0
        createGroupACK["errno"]=0;
        //错误消息为空
        createGroupACK["errmsg"]="";
        //返回成功创建的群组id
        createGroupACK["groupid"]=group.getID();

    }else{
        //创建群组失败
        //响应json错误码为1
        createGroupACK["errno"]=1;
        //错误消息
        createGroupACK["errmsg"]="create group error! ";
        //创建群组失败时默认响应群组id返回-1
        createGroupACK["groupid"]=-1;

        LOG_INFO<<"Create group failed!";
    }

    conn->send(createGroupACK.dump());

}

//加入群组业务
void ChatService::addGroup(const TcpConnectionPtr& conn,json& js,Timestamp time){
    //请求加入者id
    int userid=js["id"].get<int>();
    
    //要加入的目标群组id
    int groupid=js["groupid"].get<int>();

    //加入群组功能响应Json
    json ackJson;
    ackJson["msgid"]=int(EnMsgType::ADD_GROUP_ACK);
    ackJson["userid"]=userid;
    ackJson["groupid"]=groupid;

    //加入群组默认是普通成员类型
    if(_groupmodel.addGroup(userid,groupid,"normal")){
        //加入成功
        ackJson["errno"]=0;
        ackJson["errmsg"]="";//无错误无错误消息    
    }else{
        //加入失败
        ackJson["errno"]=1;
        ackJson["errmsg"]="User: "+std::to_string(userid)+" join group (groupid: "+std::to_string(groupid)+ ") error! ";
    }

    conn->send(ackJson.dump());
    
}

//群组聊天业务
void ChatService::groupChat(const TcpConnectionPtr& conn,json& js,Timestamp time){
    //发送者id
    int userid=js["id"].get<int>();

    //群组id
    int groupid=js["groupid"].get<int>();

    //获取该群上所有的用户
    vector<int> usersInGroup = _groupmodel.queryGroupUsers(userid,groupid);

    //遍历群成员
    for(int id:usersInGroup){
        //上锁
        std::lock_guard<std::mutex> lock(_ConnMutx);

        //如果当前遍历到的群成员在线，则直接发送消息
        if(_userConnMap.find(id)!=_userConnMap.end()){
            printf("当前服务器%d不在线！\n",id);
            //转发Json
            _userConnMap[id]->send(js.dump());
            continue;

        }else{
            // 在本服务器在线列表中找不到该用户信息的话
            // 在数据库中查询toid是否在线，从而判断是否在其它服务器中登录
            User user = _userModel.query(id);
            if (user.getState() == "online"){
                printf("数据库该用户%d在线！\n",id);
                // 该用户在其他服务器中在线
#ifdef USE_REDIS
        //在其它服务器在线的话，把消息传到redis消息队列以便达到另一服务器中
                _redis.publish(id,js.dump());
                continue;;
#endif//USE_REDIS

#ifdef USE_KAFKA
        //生产者发布消息到kafka消息队列中
                _producer->send("user"+std::to_string(id),js.dump());
                continue;
#endif
            }
            else{
                printf("数据库该用户%d不在线！，存储为离线信息\n",id);
                // 存储离线群消息
                _offlineMessageModel.insert(id, js.dump());
            }
        }
    }

}


//处理客户端注销业务
void ChatService::loginOut(const TcpConnectionPtr& conn,json& js,Timestamp time){
    int userid=js["id"].get<int>();

    {
        std::lock_guard<std::mutex> lock(_ConnMutx);
        
        //从在线用户列表中查找用户
        auto it=_userConnMap.find(userid);

        if(it!=_userConnMap.end()){
            //找到的话则将其从列表中删除
            _userConnMap.erase(userid);
        }
    }
    //用户下线，消息队列取消订阅操作
#ifdef USE_REDIS
    //用户注销相当于下线，在Redis消息队列中取消对该用户的订阅
    _redis.unsubscribe(userid);
#endif

#ifdef USE_KAFKA
    _consumer->unsubscribe("user"+std::to_string(userid));
#endif

    //更新状态为离线状态
    User user(userid,"","","offline");//因为下面函数只用到id和state，所以中间的名称密码不用理

    //调用model更新状态
    _userModel.updateState(user);

}

#ifdef USE_REDIS
//从redis消息队列上获取订阅的消息
//通道号为用户ID，可以提供查找是否在线
//Message为其它服务器上的用户发送过来的Json文本，这里原封不动发回给本机的userid
void ChatService::handleRedisSubscribeMessage(int userid, string message){
    
    //因为涉及到查找哈希表从而知道用户是否在线，所以要加锁
    std::lock_guard<std::mutex> lock(_ConnMutx);
    auto it=_userConnMap.find(userid);
    if(it!=_userConnMap.end()){
        //用户在线，直接将消息转发
        printf("REDIS 用户%d在线，直接发送消息！\n",userid);
        _userConnMap[userid]->send(message);
        return;
    }

    //不在线的话则转为离线消息
    printf("REDIS 用户%d不在线，转为离线消息！\n",userid);
    _offlineMessageModel.insert(userid,message);
}
#endif 


#ifdef USE_KAFKA
void ChatService::handleKafkaSubscribeMessage(string message){
    string temp=message;
    int userid=json::parse(temp)["toid"];
    //因为涉及到查找哈希表从而知道用户是否在线，所以要加锁
    std::lock_guard<std::mutex> lock(_ConnMutx);
    auto it=_userConnMap.find(userid);
    if(it!=_userConnMap.end()){
        //用户在线，直接将消息转发
        printf("KAFKA 用户%d在线，直接发送消息！\n",userid);
        _userConnMap[userid]->send(message);
        return;
    }

    //不在线的话则转为离线消息
    printf("KAFKA 用户%d不在线，转为离线消息！\n",userid);
    _offlineMessageModel.insert(userid,message);
}
#endif //USE_KAFKA